package com.pw.study.flink.chapter8;

import com.pw.study.flink.entities.LoginEvent;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.util.Collector;

import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.StreamSupport;

public class ExerciseLogin {


    public static void main(String[] args) {
        exercise();
    }

    private static void exercise() {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 20000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);


        //输入数据
        SingleOutputStreamOperator<LoginEvent> stream = env.readTextFile("data/file/LoginLog.csv").map(ads -> {
            String[] line = ads.split(",");
            return new LoginEvent(Long.valueOf(line[0]), line[1], line[2], Long.valueOf(line[3]) * 1000);
        });
        //


        stream.keyBy(LoginEvent::getUserId)
                .countWindow(2, 1).
                process(new ProcessWindowFunction<LoginEvent, String, Long, GlobalWindow>() {
                    @Override
                    public void process(Long key, Context ctx, Iterable<LoginEvent> elements, Collector<String> out) throws Exception {
                        List<LoginEvent> list = StreamSupport.stream(elements.spliterator(), true).collect(Collectors.toList());
                        if (list.size() == 2) {
                            LoginEvent event1 = list.get(0);
                            LoginEvent event2 = list.get(1);
                            if (event1.getEventType().equals("fail") && event2.getEventType().equals("fail") && Math.abs(event1.getEventTime() - event2.getEventTime()) < 2000) {
                                out.collect("用户: " + key + " 在2s内在进行恶意登录");
                            }
                        }
                    }
                }).print();


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}
